package com.cn.producer.impl;

import com.cn.client.MqFaultStrategy;
import com.cn.client.TopicPublishInfo;
import com.cn.common.*;
import com.cn.common.processor.ProcessorCode;
import com.cn.exception.MyException;
import com.cn.file.AppendMessageResult;
import com.cn.file.MessageStore;
import com.cn.msg.Message;
import com.cn.producer.Producer;
import com.cn.producer.SendResult;
import com.cn.producer.SendStatus;
import com.cn.remote.RemoteClient;
import com.cn.remote.RemoteClient2;
import com.cn.util.RemotingSerializable;
import com.cn.util.SerializationUtil;

import java.nio.charset.Charset;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeMap;
import java.util.concurrent.*;

/**
 * 发送消息
 */
public class ProducerImpl implements Producer {

    private String nameSrv;

    private final String bodys = "default";

    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private ConcurrentMap<String /*brokerName*/, TreeMap<Integer, String> /*broker节点集*/> brokerNodes = new ConcurrentHashMap<>();

    private ConcurrentMap<String /*topic*/, List<QueueData>> topicTables = new ConcurrentHashMap<>();

    private ConcurrentMap<String /*topic*/, TopicPublishInfo> topicPublishTables = new ConcurrentHashMap<>();

    private ConcurrentMap<String /* topic */, TopicRouteData> topicRouteTables = new ConcurrentHashMap<>();

    private RemoteClient2 client;
    private MqFaultStrategy mqFaultStrategy;

    public ProducerImpl(String nameSrv) {
        this.nameSrv = nameSrv;
        client = new RemoteClient2(null);
        mqFaultStrategy = new MqFaultStrategy();
    }

    @Override
    public void start() throws InterruptedException {
        this.client.start();
        this.schuleTask();
    }

    @Override
    public void stop() {
        this.client.shutdown();
    }

    public void schuleTask() {

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                pullTopic();
            }
        }, 5, 5, TimeUnit.SECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                for (String key : topicRouteTables.keySet()) {
                    pullTopic(key);
                }
            }
        }, 30, 30, TimeUnit.SECONDS);
    }

    public void pullTopic(String topic) {
        Message message = new Message();
        message.setType((byte) 15);
        message.setFlag((byte) 1);
        message.setByteBodys(topic.getBytes(Charset.forName("utf-8")));
        message.setLength(message.getByteBodys().length);
        message.setBody(null);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_default);
        try {
            byte[] result = this.client.invokeMsg(message, this.nameSrv);
            if (result != null) {
                TopicRouteData topicRouteData = RemotingSerializable.decode(result, TopicRouteData.class);
                if (topicRouteData == null) {
                    return;
                }

                TopicRouteData old = this.topicRouteTables.get(topic);
                if (null == old) {
                    old = new TopicRouteData();
                    old = this.topicRouteTables.putIfAbsent(topic, old);
                }

                boolean change = false;
                if (!old.equals(topicRouteData)) {
                    change = true;
                }
                if (change) {
                    TopicPublishInfo topicPublishInfo = this.topicPublishTables.get(topic);
                    if (topicPublishInfo == null) {
                        topicPublishInfo = new TopicPublishInfo();
                        topicPublishInfo = this.topicPublishTables.putIfAbsent(topic, topicPublishInfo);
                    }
                    topicPublishInfo.setTopicRouteData(topicRouteData);

                    List<MessageQueue> messageQueueList = new ArrayList<>();
                    for (QueueData queueData : topicRouteData.getQueueData()) {
                        for (int i = 0; i < queueData.getWriteQueue(); i++) {
                            MessageQueue messageQueue = new MessageQueue();
                            messageQueue.setBrokerName(queueData.getBrokerName()).setQueueId(i).setTopic(topic);
                            messageQueueList.add(messageQueue);
                        }
                    }
                    topicPublishInfo.setMessageQueues(messageQueueList);
                    cacheBroker(topicRouteData.getBrokerData());
                }

            }

        } catch (Exception e) {

        }
    }

    public void pullTopic() {
        Message message = new Message();
        message.setType((byte) 19);
        message.setFlag((byte) 1);
        message.setByteBodys(bodys.getBytes(Charset.forName("utf-8")));
        message.setLength(message.getByteBodys().length);
        message.setBody(null);
        message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_default);
        try {
            byte[] result = this.client.invokeMsg(message, this.nameSrv);
            if (result != null) {
                TopicRouteData topicRouteData = RemotingSerializable.decode(result, TopicRouteData.class);
                if (topicRouteData == null) {
                    return;
                }

                ConcurrentMap<String, List<QueueData>> tmp = new ConcurrentHashMap<>();
                for (QueueData queueData : topicRouteData.getQueueData()) {
                    List<QueueData> queueDataList = tmp.get(queueData.getTopic());
                    if (queueDataList == null) {
                        queueDataList = new ArrayList<>();
                        queueDataList.add(queueData);
                    } else {
                        queueDataList.add(queueData);
                    }
                    tmp.put(queueData.getTopic(), queueDataList);
                }

                for (String key : tmp.keySet()) {
                    List<QueueData> queueDataList = this.topicTables.get(key);
                    if (queueDataList == null) {
                        queueDataList = tmp.get(key);
                        this.topicTables.putIfAbsent(key, queueDataList);
                    }
                    if (!queueDataList.equals(tmp.get(key))) {
                        this.topicTables.put(key, tmp.get(key));
                    }
                }
                cacheBroker(topicRouteData.getBrokerData());
            }
        } catch (Exception e) {

        }
    }

    public void cacheBroker(List<BrokerData> brokerDataList) {
        for (BrokerData brokerData : brokerDataList) {
            TreeMap<Integer, String> old = this.brokerNodes.get(brokerData.getBrokerName());
            if (old == null) {
                old = brokerData.getAddr();
                this.brokerNodes.putIfAbsent(brokerData.getBrokerName(), old);
            }
            if (!old.equals(brokerData.getAddr())) {
                this.brokerNodes.put(brokerData.getBrokerName(), brokerData.getAddr());
            }
        }
    }

    @Override
    public SendResult send(MessageStore messageStore) {
        SendResult result = new SendResult();

        TopicPublishInfo topicPublishInfo = selectTopic(messageStore.getTopic());
        if (topicPublishInfo == null) {
            //抛出或者打印异常
            System.out.println("没有拉取到broker信息！");
        } else {
            String brokerName = null;
            for (int i = 0; i < 1; i++) {
                MessageQueue messageQueue = selectMessageQueue(topicPublishInfo, false, brokerName);
                if (messageQueue != null) {
                    brokerName = messageQueue.getBrokerName();
                    String addr = findBrokerAddressInPublish(messageQueue.getBrokerName());

                    messageStore.setQueueId(messageQueue.getQueueId());
                    Message message = new Message();
                    message.setFlag((byte) 2);
                    message.setType((byte) 8);
                    message.setNettyRequestWrapperCode((byte) ProcessorCode.push_msg);
                    message.setByteBodys(RemotingSerializable.encode(messageStore));
                    message.setLength(message.getByteBodys().length);
                    message.setBody(null);
                    try {
                        byte[] bytes = this.client.invokeMsg(message, addr);
                        if (bytes != null) {
                            AppendMessageResult messageResult = RemotingSerializable.decode(bytes, AppendMessageResult.class);
                            switch (messageResult.getStatus()) {
                                case PUT_OK:
                                    result.setSendStatus(SendStatus.SEND_OK);
                                    break;
                                default:
                                    result.setSendStatus(SendStatus.SEND_FAIL);
                                    break;
                            }
                        } else {
                            result.setSendStatus(SendStatus.SEND_FAIL);
                        }
                        return result;
                    } catch (Exception e) {
                        e.printStackTrace();

                    }
                }
            }

        }

        return result;
    }

    public String findBrokerAddressInPublish(String brokerName) {
        TreeMap<Integer, String> treeMap = this.brokerNodes.get(brokerName);
        if (null != treeMap) {
            return treeMap.get(0);
        }
        return null;
    }

    public MessageQueue selectMessageQueue(TopicPublishInfo topicPublishInfo, boolean fair, String brokerName) {
        if (!fair) {
            return topicPublishInfo.selectMessageQueue(brokerName);
        } else {
            //
            return mqFaultStrategy.selectMessageQueue(topicPublishInfo, brokerName);
        }
    }

    public TopicPublishInfo selectTopic(String topic) {
        System.out.println(topic + ">>>>>>>>");
        TopicPublishInfo topicPublishInfo = this.topicPublishTables.get(topic);
        if (topicPublishInfo == null) {
            pullTopic(topic);
            topicPublishInfo = this.topicPublishTables.get(topic);
        }
        return topicPublishInfo;
    }

    @Override
    public SendResult send(MessageStore messageStore, long timeout) {
        return null;
    }

    @Override
    public SendResult invokeMsg(Message message, String addr) {
        return null;
    }

    @Override
    public void updateTopicPublishInfo(String topic, TopicPublishInfo topicPublishInfo) {
        this.topicPublishTables.put(topic, topicPublishInfo);
    }
}
